package com.ku.rocketmq.quickstart.orderconsume;

import com.ku.rocketmq.constants.Const;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * @program: rocketmqDemo
 * @description: 顺序消费消息 --- 生产者
 * @author: JokerKu
 * @create: 2020-05-08 10:05
 **/
public class OrderProvider {

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {

        DefaultMQProducer producer = new DefaultMQProducer("orderConsume_provider_group");
        producer.setNamesrvAddr(Const.NAMESERVER_ADDR);
        producer.start();

        Date date = new Date();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String currentTime = sdf.format(date);

        //这五条消息是一个业务操作，要按个按照顺序来消费
        for (int i = 0; i < 5; i++) {
            //时间戳
            String body = currentTime+"Hello MQ"+i;
            //封装消息
            Message message = new Message("test_order_topic","TagA","Key"+i,body.getBytes());
            //发送数据
            //如果使用顺序消费，则必须自己实现MessageQueueSelector,保证消息进入同一队列
            SendResult sendResult = producer.send(message, (list, message1, arg) -> {
                Integer queueId = (Integer) arg;
                return list.get(queueId);
            }, 1);//1是队列下标
            System.out.println("sendResult:"+ sendResult+"----body:"+body);
        }

        //这五条消息是一个业务操作，要按个按照顺序来消费
        for (int i = 0; i < 5; i++) {
            //时间戳
            String body = currentTime+"Hello MQ"+i;
            //封装消息
            Message message = new Message("test_order_topic","TagA","Key"+i,body.getBytes());
            //发送数据
            //如果使用顺序消费，则必须自己实现MessageQueueSelector,保证消息进入同一队列
            SendResult sendResult = producer.send(message, (list, message1, arg) -> {
                Integer queueId = (Integer) arg;
                return list.get(queueId);
            }, 2);//2是队列下标
            System.out.println("sendResult:"+ sendResult+"----body:"+body);
        }

        //这五条消息是一个业务操作，要按个按照顺序来消费
        for (int i = 0; i < 5; i++) {
            //时间戳
            String body = currentTime+"Hello MQ"+i;
            //封装消息
            Message message = new Message("test_order_topic","TagA","Key"+i,body.getBytes());
            //发送数据
            //如果使用顺序消费，则必须自己实现MessageQueueSelector,保证消息进入同一队列
            SendResult sendResult = producer.send(message, (list, message1, arg) -> {
                Integer queueId = (Integer) arg;
                return list.get(queueId);
            }, 3);//3是队列下标
            System.out.println("sendResult:"+ sendResult+"----body:"+body);
        }

        producer.shutdown();
    }
}
